package redis

import (
	"corelib/log"
	"fmt"
	"github.com/go-redis/redis/v8"
	"runtime/debug"
	"time"
)

type Producer struct {
	rds    *RedisIns
	maxLen int64
}

type SteamOption struct {
	ProduceMsgMaxLen int64
	ProduceApprox    bool // stream队列最大数量是否弹性

	ConsumeMsgStId   string
	ConsumeReadBlock time.Duration
	ReadCountEach    int64

	GroupConsumeNoAck bool
}

type SteamOptionSet func(options *SteamOption)

var (
	_streamOpt = &SteamOption{
		ProduceMsgMaxLen:  _produceMsgMaxLen,
		ConsumeMsgStId:    _comsumeGroupMsgNew,
		ConsumeReadBlock:  _comsumeReadBlock,
		ReadCountEach:     5,
		GroupConsumeNoAck: true,
		ProduceApprox:     true,
	}
)

const (
	_produceMsgMaxLen = 5000

	_comsumeGroupMsgNew = ">" // > means give me new message when group consume
	_comsumeMsgNew      = "$" // > means give me new message;
	_comsumeMsgNow      = "0" // 0 means give me exists message

	_comsumeReadBlock = 500 * time.Millisecond
)

func SetStreamOption(o ...SteamOptionSet) {
	for _, set := range o {
		set(_streamOpt)
	}
}

func NewProducerByRdsConf(c *RdsConfig, maxLen ...int64) (*Producer, error) {
	if c.Name == "" {
		c.Name = "streamProducer"
	}
	r, e := NewRedisIns(c)
	if e != nil {
		return nil, e
	}

	p := &Producer{
		rds:    r,
		maxLen: _streamOpt.ProduceMsgMaxLen,
	}
	if len(maxLen) > 0 && maxLen[0] > 0 {
		p.maxLen = maxLen[0]
	}
	return p, nil
}

func NewProducer(rdsIdx string, maxLen ...int64) (*Producer, error) {
	if rdsIdx == "" {
		rdsIdx = "default"
	}
	r, e := rdsMap[rdsIdx]
	if !e {
		return nil, fmt.Errorf("redis %v was not exists", rdsIdx)
	}
	p := &Producer{
		rds:    r,
		maxLen: 10000,
	}
	if len(maxLen) > 0 && maxLen[0] > 0 {
		p.maxLen = maxLen[0]
	}
	return p, nil
}

func (p *Producer) Send(topic string, data interface{}) (string, error) {
	return p.rds.ins.XAdd(p.rds.ctx, &redis.XAddArgs{
		Stream: topic,
		MaxLen: p.maxLen,
		Approx: _streamOpt.ProduceApprox,
		Values: data,
	}).Result()
}

func NewGroupConsumer(rds *RedisIns, group, consumer string, topic []string, async bool, cbk func(topic, id string, value map[string]interface{}), stopCh chan bool) error {
	// 创建消费者组
	streams := make([]string, len(topic))
	for i, s := range topic {
		e := rds.ins.XGroupCreateMkStream(rds.ctx, s, group, _comsumeMsgNow).Err()
		if e != nil {
			return e
		}
		streams[i] = _comsumeMsgNew
	}
	// 开始启动消费协程
	streams = append(topic, streams...)
	doConsume := func() {
		defer func() {
			if p := recover(); p != nil {
				log.Printf("group consumer %s panic: %v\nstack:%s", consumer, p, string(debug.Stack()))
			}
		}()
		msg, err := rds.ins.XReadGroup(rds.ctx, &redis.XReadGroupArgs{
			Group:    group,
			Consumer: consumer,
			Streams:  streams,
			Count:    _streamOpt.ReadCountEach,
			Block:    _streamOpt.ConsumeReadBlock,
			NoAck:    _streamOpt.GroupConsumeNoAck,
		}).Result()
		if err != nil && err != redis.Nil {
			log.Printf("stream XReadGroup err:%v", err)
		} else {
			for _, xStream := range msg {
				for _, one := range xStream.Messages {
					if !_streamOpt.GroupConsumeNoAck {
						rds.ins.XAck(rds.ctx, xStream.Stream, group, one.ID)
					}
					if async {
						go cbk(xStream.Stream, one.ID, one.Values)
					} else {
						cbk(xStream.Stream, one.ID, one.Values)
					}
				}

			}
		}
	}
	for {
		select {
		case <-stopCh:
			break
		default:
			doConsume()
		}
	}
	return nil
}

func NewConsumer(rds *RedisIns, topic []string, async bool, cbk func(topic, id string, value map[string]interface{}), stopCh chan bool) error {
	// 填充消费主体的起始id
	streams := make([]string, len(topic))
	for i, _ := range topic {
		streams[i] = _comsumeMsgNew
	}
	// 开始启动消费协程
	streams = append(topic, streams...)
	doConsume := func() {
		defer func() {
			if p := recover(); p != nil {
				log.Printf("consumer panic: %v\nstack:%s", p, string(debug.Stack()))
			}
		}()
		msg, err := rds.ins.XRead(rds.ctx, &redis.XReadArgs{
			Streams: streams,
			Count:   _streamOpt.ReadCountEach,
			Block:   _streamOpt.ConsumeReadBlock,
		}).Result()
		if err != redis.Nil && err != nil {
			log.Printf("stream XRead err:%v", err)
		} else {
			for _, xStream := range msg {
				for _, one := range xStream.Messages {
					if async {
						go cbk(xStream.Stream, one.ID, one.Values)
					} else {
						cbk(xStream.Stream, one.ID, one.Values)
					}
				}

			}
		}
	}
	for {
		select {
		case <-stopCh:
			break
		default:
			doConsume()
		}
	}
	return nil
}
